package net.alan.etl;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.log4j.Logger;

import net.alan.etl.i.IFlowTask;
import net.alan.etl.kafka.FlowParams;
import net.alan.etl.proxy.FlowProxyFactory;
import net.dotool.thread.ThreadPool;

/**
 * 
 * @author 李岩飞
 * @email eliyanfei@126.com
 * @date 2018年5月10日下午2:19:12
 */
public class FlowConsumer implements Runnable {
	static Logger logger = Logger.getLogger(FlowConsumer.class);
	private Map<String, IFlowTask> taskMap = new HashMap<String, IFlowTask>();
	private ThreadPool<Runnable> taskPools = new ThreadPool<Runnable>(20);

	/**
	 * 注册
	 */
	public void subscribe(String topic, IFlowTask flowTask) {
		taskMap.put(topic, flowTask);
	}

	KafkaConsumer<String, FlowParams> consumer;
	Properties props = new Properties();

	public FlowConsumer() {
	}

	public void initConsumer() {
		props.put("bootstrap.servers", "localhost:9092");
		props.put("group.id", "1");
		props.put("enable.auto.commit", "true");
		props.put("auto.commit.interval.ms", "1000");
		// 设置使用最开始的offset偏移量为该group.id的最早。如果不设置，则会是latest即该topic最新一个消息的offset
		// 如果采用latest，消费者只能得道其启动后，生产者生产的消息
		props.put("auto.offset.reset", "earliest");
		props.put("session.timeout.ms", "30000");
		props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		props.put("value.deserializer", "net.alan.etl.kafka.FlowParamDecodeing");
		consumer = new KafkaConsumer<>(props);
		// 订阅topic
		consumer.subscribe(taskMap.keySet());

		logger.warn("Flow 消费者已启动...");
	}

	@Override
	public void run() {
		initConsumer();
		while (true) {
			try {
				ConsumerRecords<String, FlowParams> records = consumer.poll(10);
				for (ConsumerRecord<String, FlowParams> record : records) {
					startTask(record.topic(), record.value());
				}
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
	}

	/**
	 * 启动任务
	 */
	public void startTask(String topic, FlowParams params) {
		IFlowTask task = (IFlowTask) new FlowProxyFactory(taskMap.get(topic)).getFlowTask();
		task.initParam(params);
		taskPools.execute(task);
	}

	/**
	 * 启动消费者
	 */
	public void startup() {
		new Thread(this).start();
	}

	@Override
	protected Object clone() throws CloneNotSupportedException {
		try {
			this.consumer.close();
		} catch (Exception e) {
		}
		return super.clone();
	}
}